package job

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/elastic/go-elasticsearch/v8"
	"github.com/go-kratos/kratos/v2/log"
	"github.com/segmentio/kafka-go"
	"review-job/internal/conf"
)

type Msg struct {
	Type     string `json:"type"`
	database string `json:"database"`
	table    string `json:"table"`
	isDel    bool   `json:"isDel"`
	Data     []map[string]interface{}
}

type JobWorker struct {
	kafkaReader *kafka.Reader
	esClient    *ESClient
	log         *log.Helper
}
type ESClient struct {
	*elasticsearch.TypedClient
	index string
}

func NewJobWorker(kafka *kafka.Reader, esClient *ESClient, logger log.Logger) *JobWorker {
	return &JobWorker{
		kafkaReader: kafka,
		esClient:    esClient,
		log:         log.NewHelper(logger),
	}
}

func NewKafka(cfg *conf.Kafka) *kafka.Reader {
	return kafka.NewReader(kafka.ReaderConfig{
		Brokers: cfg.Brokers,
		GroupID: cfg.GroupId, // 指定消费者组id
		Topic:   cfg.Topic,
	})
}

func NewESClient(cfg *conf.Elasticsearch) (*ESClient, error) {
	c := elasticsearch.Config{
		Addresses: cfg.Addresses,
	}
	client, err := elasticsearch.NewTypedClient(c)
	if err != nil {
		fmt.Printf("elasticsearch.NewTypedClient failed, err:%v\n", err)
		return nil, err
	}
	return &ESClient{
		TypedClient: client,
		index:       cfg.Index,
	}, nil
}

func (jw JobWorker) Start(ctx context.Context) error {
	// kafka接收消息
	for {
		m, err := jw.kafkaReader.ReadMessage(ctx)
		if errors.Is(err, context.Canceled) {
			return nil
		}
		if err != nil {
			jw.log.Errorf("kafka read message failed: %v", err)
			break
		}
		jw.log.Errorf("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))
		//将数据读取到ES
		msg := new(Msg)
		err = json.Unmarshal(m.Value, msg)
		if err != nil {
			jw.log.Errorf("unmarshal message form kafka failed: %v", err)
			continue
		}

		if msg.Type == "INSERT" {
			//新增文档
			for idx := range msg.Data {
				jw.indexDocument(msg.Data[idx])
			}

		} else {
			//更新文档
			for idx := range msg.Data {
				jw.updateDocument(msg.Data[idx])
			}
		}

	}

	return nil
}
func (jw JobWorker) Stop(context.Context) error {
	if err := jw.kafkaReader.Close(); err != nil {
		log.Fatal("failed to close reader:", err)
	}

	return nil
}

// indexDocument 索引文档
func (jw JobWorker) indexDocument(d map[string]interface{}) {
	// 添加文档
	reviewID := d["review_id"].(string)
	resp, err := jw.esClient.Index(jw.esClient.index).
		Id(reviewID).
		Document(d).
		Do(context.Background())
	if err != nil {
		jw.log.Errorf("indexing document failed, err:%v\n", err)
		return
	}
	jw.log.Debugf("result:%#v\n", resp.Result)
}

// updateDocument 更新文档
func (jw JobWorker) updateDocument(d map[string]interface{}) {
	reviewID := d["review_id"].(string)
	resp, err := jw.esClient.Update("my-review-1", "1").
		Doc(reviewID). // 使用结构体变量更新
		Do(context.Background())
	if err != nil {
		jw.log.Errorf("update document failed, err:%v\n", err)
		return
	}
	jw.log.Debugf("result:%v\n", resp.Result)
}
